/*
 * Copyright (c) 2012-2014 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package org.dna.mqtt.moquette.processor.impl;

import org.dna.mqtt.moquette.messaging.spi.impl.ValueEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.ProtocolEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.processor.ProtocolProcessor;
import org.dna.mqtt.moquette.processor.ProtocolProcessorFactory;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.ServerChannel;

/**
 * @author puran
 */
public class MqttProtocolProcessorFactory implements ProtocolProcessorFactory {
	
	@Override
	public ProtocolProcessor getProcessor(ValueEvent valueEvent, MqttProtocolProcessor baseProcessor) {
		MessagingEvent evt = valueEvent.getEvent();
		
		ProtocolProcessor protocolProcessor = null;
		
		if (evt instanceof PublishEvent) {
			protocolProcessor = new PublishProcessor((PublishEvent) evt);
        } else if (evt instanceof DisconnectEvent) {
            DisconnectEvent disconnectEvent = (DisconnectEvent) evt;
            disconnectEvent.getSession().setAttribute(Constants.ATTR_CLIENTID, false);
            protocolProcessor = new DisconnectProcessor(disconnectEvent.getSession());
        } else if (evt instanceof ProtocolEvent) {
        	protocolProcessor = createProtocolProcessor(evt);
        } 
		
		if (protocolProcessor != null) {
			protocolProcessor.setBaseProcessor(baseProcessor);
		}
		return protocolProcessor;
	}

	private ProtocolProcessor createProtocolProcessor(MessagingEvent evt) {
		 ServerChannel session = ((ProtocolEvent) evt).getSession();
         AbstractMessage message = ((ProtocolEvent) evt).getMessage();
         
         if (message instanceof ConnectMessage) {
             return new ConnectProcessor(session, (ConnectMessage) message);
         } else if (message instanceof SubscribeMessage) {
        	 return new SubscribeProcessor(session, (SubscribeMessage)message);
         } else if (message instanceof UnsubscribeMessage) {
             return new UnSubscribeProcessor(session, (UnsubscribeMessage) message);
         } else if (message instanceof  PublishMessage) {
             PublishEvent publishEvent = new PublishEvent((PublishMessage) message, session.getClientID());
             return new PublishProcessor(publishEvent);
         } else if (message instanceof DisconnectMessage) {
        	 return new DisconnectProcessor(session); 
         } else if (message instanceof PubRelMessage) {
             return new PubRelProcessor(session, (PubRelMessage) message);
         } else if (message instanceof PubRecMessage) {
        	 return new PubRecProcessor(session, (PubRecMessage) message);
         } else if (message instanceof PubCompMessage) {
             return new PubCompProcessor(session, (PubCompMessage) message);
         } else if (message instanceof PubAckMessage) {
             return new PubAckProcessor(session, (PubAckMessage) message);
         }
         return null;
	}
}
